﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Messaging;
using System.ServiceModel;
using System.Threading.Tasks;
using TestService.ApplicationService.Data;
using TestService.ApplicationService.Services.Interface;
using TestService.Common.IoC;
using TestService.Contract;
using TestService.Contract.Data;
using TestService.Contract.Faults;

namespace TestService.ApplicationService
{
    [IocServiceBehavior]
    [ServiceBehavior(InstanceContextMode = InstanceContextMode.Single, ConcurrencyMode = ConcurrencyMode.Multiple)]
    public class EventBroker : ApplicationBaseService, IEventBroker
    {
        Dictionary<string, List<UniqueCallbackHandle>> eventNameToCallbackLookups = new Dictionary<string, List<UniqueCallbackHandle>>();
        private static Object syncObj = new Object();
        private static string inputQueueName = "";
        private bool shouldRun = true;
        private static readonly TimeSpan queueReadTimeOut = TimeSpan.FromSeconds(500);
        private static readonly TimeSpan queuePeekTimeOut = TimeSpan.FromSeconds(30);
        private IXmlParserService _xmlParserService;

        public EventBroker(IXmlParserService xmlParserService)
        {
            inputQueueName = AppSettings.InputQueueName;
            StartCollectingMessage();
            _xmlParserService = xmlParserService;
        }

        public void StartCollectingMessage()
        {
            try
            {
                GetMessageFromQueue();
            }
            catch (Exception ex)
            {
                throw new FaultException<EventBrokerException>(new EventBrokerException(ex.Message), new FaultReason(ex.Message));
            }
        }

        public void Subscribe(Guid subscriptionId, string[] eventNames)
        {
            try
            {
                CreateSubscription(subscriptionId, eventNames);
            }
            catch (Exception ex)
            {
                throw new FaultException<EventBrokerException>(new EventBrokerException(ex.Message), new FaultReason(ex.Message));
            }
        }

        public void EndSubscription(Guid subscriptionId)
        {
            lock (syncObj)
            {
                //create new dictionary that will be populated by those remaining
                Dictionary<string, List<UniqueCallbackHandle>> remainingEventNameToCallbackLookups =
                    new Dictionary<string, List<UniqueCallbackHandle>>();

                foreach (KeyValuePair<string, List<UniqueCallbackHandle>> kvp in eventNameToCallbackLookups)
                {
                    //get all the remaining subscribers whos session id is not the same as the one we wish to remove
                    List<UniqueCallbackHandle> remainingMessageSubscriptions =
                        kvp.Value.Where(x => x.CallbackSessionId != subscriptionId).ToList();
                    if (remainingMessageSubscriptions.Any())
                    {
                        remainingEventNameToCallbackLookups.Add(kvp.Key, remainingMessageSubscriptions);
                    }
                }
                //now left with only the subscribers that are subscribed
                eventNameToCallbackLookups = remainingEventNameToCallbackLookups;
            }
        }

        #region 私有方法

        /// <summary>
        /// 从消息队列中获取消息
        /// </summary>
        private void GetMessageFromQueue()
        {
            try
            {
                Task messageQueueReaderTask = Task.Factory.StartNew(() =>
                {
                    using (MessageQueue queue = new MessageQueue(inputQueueName, QueueAccessMode.Receive))
                    {

                        queue.Formatter = new XmlMessageFormatter(new[] { typeof(string) });

                        while (shouldRun)
                        {
                            Message message = null;

                            try
                            {
                                if (!queue.IsEmpty())
                                {
                                    //Logs.Debug("接受队列里的消息");
                                    message = queue.Receive(queueReadTimeOut);
                                    ProcessMessage(message);
                                }
                            }
                            catch (MessageQueueException e)
                            {
                                if (e.MessageQueueErrorCode != MessageQueueErrorCode.IOTimeout)
                                {
                                    Log.Warn("消息队列出现异常：", e);
                                }
                            }
                            catch (Exception e)
                            {
                                // Write the message details to the Error queue
                                Log.Warn("操作异常：", e);
                            }
                        }
                    }
                }, TaskCreationOptions.LongRunning);
            }
            catch (AggregateException ex)
            {
                throw;
            }
        }

        /// <summary>
        /// 处理消息
        /// </summary>
        /// <param name="msmqMessage"></param>
        private void ProcessMessage(Message msmqMessage)
        {
            string messageBody = (string)msmqMessage.Body;

#if DEBUG
            Log.Info(string.Format("接受消息 : {0}", messageBody));
#endif

            RealTimeEventMessage messageToSendToSubscribers = _xmlParserService.ParseRawMsmqXml(messageBody);
            if (messageToSendToSubscribers != null)
            {
                lock (syncObj)
                {
                    if (messageToSendToSubscribers.Msg != null)
                    {
                        // 保存到数据库
                        
                    }

                    List<Guid> deadSubscribers = new List<Guid>();

                    if (eventNameToCallbackLookups.ContainsKey(messageToSendToSubscribers.EventName))
                    {
                        List<UniqueCallbackHandle> uniqueCallbackHandles =
                            eventNameToCallbackLookups[messageToSendToSubscribers.EventName];
                        foreach (UniqueCallbackHandle uniqueCallbackHandle in uniqueCallbackHandles)
                        {
                            try
                            {
                                uniqueCallbackHandle.Callback.ReceiveStreamingResult(messageToSendToSubscribers);

                            }
                            catch (CommunicationObjectAbortedException coaex)
                            {
                                deadSubscribers.Add(uniqueCallbackHandle.CallbackSessionId);
                            }
                        }
                    }

                    //end all subcriptions for dead subscribers
                    foreach (Guid deadSubscriberId in deadSubscribers)
                    {
                        EndSubscription(deadSubscriberId);
                    }
                }
            }
        }

        private void CreateSubscription(Guid subscriptionId, string[] eventNames)
        {
            //Ensure that a subscription is created for each message type the subscriber wants to receive
            lock (syncObj)
            {
                foreach (string eventName in eventNames)
                {
                    if (!eventNameToCallbackLookups.ContainsKey(eventName))
                    {
                        List<UniqueCallbackHandle> currentCallbacks = new List<UniqueCallbackHandle>();
                        eventNameToCallbackLookups[eventName] = currentCallbacks;
                    }
                    eventNameToCallbackLookups[eventName].Add(
                        new UniqueCallbackHandle(subscriptionId, OperationContext.Current.GetCallbackChannel<IEventBrokerCallback>()));
                }
            }
        }

        #endregion
    }
}
